package com.atguigu.bigdata.chapter05.transform;

import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * @Author lzc
 * @Date 2022/9/2 10:26
 */
public class Flink05_KeyBy_1 {
    public static void main(String[] args) {
        Configuration conf = new Configuration();
        conf.setInteger("rest.port", 2000);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
        env.setParallelism(2);
        
        
        // 分别计算奇数的和 与 偶数的和
        // 要聚合: 必须先分组(keyBy), 聚合的时候就是每组内实现聚合
        env
            .fromElements(1, 10, 3, 6, 9)
            .keyBy(new KeySelector<Integer, String>() {
                @Override
                public String getKey(Integer value) throws Exception {
                    return "";
                }
            })
            .sum(0)
            .print();
        
        
        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}
/*
什么类型可以做为key?
   实际使用, 一般使用String作为key, (其他的基本类型, 去重浮点类型)
   
其实是对key的双重hash然后分组,
    同一组(key一样), 则一定在同一个并行度中
    
    不同组(key不一样)的值, 可能在同一个并行度,也可能比在
    
    
如何根据key选择并行度:
new KeyedStream<>(this, clean(key));



new KeyGroupStreamPartitioner<>(
                                keySelector,
                                128))
                                

"奇数"/"偶数", 128, 2
KeyGroupRangeAssignment.assignKeyToParallelOperator(
                key, maxParallelism, numberOfChannels);
                
128, 2,  [0, 128)
return computeOperatorIndexForKeyGroup(
                maxParallelism, parallelism, assignToKeyGroup(key, maxParallelism));
          
                
       "奇数"/"偶数",128   本质:对key两次hash之后对128取余
       assignToKeyGroup(key, maxParallelism)
            // "奇数"/"偶数" 的hashCode值, 128   // 对key第一次hash
            computeKeyGroupForKeyHash(key.hashCode(), maxParallelism);
                MathUtils.murmurHash(keyHash) % maxParallelism;  // 对key进行第二次hash
  [0,128) * 2 / 128
  keyGroupId * parallelism / maxParallelism
  

 并行度索引 =  [0,128) * 2 / 128
   [0, 64) 索引是 0
   
   [64, 128) 索引 1
   
   换句话说: 如果key经过双重hash% 128 之后, 在64的同一次, 则在同一个并行度中.
   
            
 */